#include <ILogger.h>
#include "DomainServlet.h"

#include <stdio.h> /* for printf() */
#include <sys/types.h> /* for Socket data types */
#include <sys/socket.h> /* for socket(), connect(), send(), and recv() */
#include <sys/un.h> /* for sockaddr_un */
#include <netinet/in.h> /* for IP Socket data types */
#include <string.h> /* for memset() */
#include <errno.h> /* for errno*/
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>
#include <thread>
#include <time.h>
#include <sys/time.h>

#define BUF_SIZE 1024


class SocketSession : public ISession {
public:
    int socket;
    SocketSession(int socket) :socket(socket){

    }
    void reply(const void *data,int len)
    {
        send(socket,data,len,0);
    }
    void close(){

    }
};

void DomainClientServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler)
{
    msgidHandlers[type].push_back(handler);
}
void DomainClientServlet::request(const void *data,int len)
{
    int i = send(conn_fd, data, len, 0);
    LogI("DomainClientServlet::request want:%d real:%d\n",len,i);
}

void DomainClientServlet::recvthreadLogic(int socket) {
    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    SocketSession session(socket);
    onState(OPEN,&session,0);
    
    while (isRun) {
        char buf[BUF_SIZE];
        memset(buf, 0, BUF_SIZE);
        int n = recv(socket, buf, BUF_SIZE, 0);
         if(n == -1) { //recv函数返回错误

            if (errno == EAGAIN) {
                continue;
            }
            onState(ERROR,&session,0);
            LogI("client recv error\n");
            close(socket);
            socket = -1;
            break;
        } else if(n == 0) { //client端套接字被关闭
            LogI("xx client is closed\n");
            onState(CLOSE,&session,0);
            close(socket);
            socket = -1;
            break;
        } else { //读取成功
            /*对收到的数据进行处理*/
            LogI("client receive data:%d\n", n);
            
            onMessage(&session,buf,n);
        }
    }
    LogI("DomainClientServlet::recvthreadLogic end\n");
}
void DomainClientServlet::threadLogic()
{
    struct sockaddr_un server_addr;

    /*创建unix域套接字*/
    conn_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if(conn_fd == -1)
    {
        LogI("socket create error\n");
        return;
    }     

    /*指定将连接的server信息*/
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sun_family = AF_UNIX;
    strcpy(server_addr.sun_path, url.c_str());

    /*连接到server*/
    int retryCount = 0;
    while(connect(conn_fd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1 && retryCount++ < 3)
    {
        LogI("connect error retryCount:%d",retryCount);
        std::this_thread::sleep_for(std::chrono::milliseconds(1500));
    }

    auto client = std::thread(&DomainClientServlet::recvthreadLogic,this,conn_fd);
    LogI("connect to server on path:%s", url.c_str());
    
   
    /*关闭描述符*/
    LogI("try to close client");
    client.join();
    close(conn_fd);
    LogI("client end\n");
}
void DomainClientServlet::start()
{
    // create a thread handle things in a thread
    _thread = new std::thread(&DomainClientServlet::threadLogic,this);
}
void DomainClientServlet::wait()
{
    ((std::thread*)_thread)->join();
    
}
void DomainClientServlet::stop()
{
    isRun = false;
}
void DomainClientServlet::onLoad(INode *node)
{
    INode *child = node->childNode("connection");
    if (child) {
        name = child->getStr("name");
        url = child->getStr("url");
    }
    start();
}
void DomainClientServlet::onMessage(ISession *session,void* data,int size)
{
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void DomainClientServlet::onState(SERVLET_STATUS status,ISession *session,int code)
{
    auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
         for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}

void DomainClientServlet::onUnload()
{
    stop();
    wait();
}
DomainClientServlet::~DomainClientServlet()
{
     delete (std::thread*)_thread; 
}

void DomainServerServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler)
{
    msgidHandlers[type].push_back(handler);
}
void DomainServerServlet::request(const void *data,int len)
{
    (void)data;
    (void)len;
}
void DomainServerServlet::recvthreadLogic(int socket)
{
    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    SocketSession session(socket);
    onState(OPEN,&session,0);
    while (isRun) {
        char buf[BUF_SIZE];
        memset(buf, 0, BUF_SIZE);
        int n = recv(socket, buf, BUF_SIZE, 0);
         if(n == -1) { //recv函数返回错误
            
            if (errno == EAGAIN) {
                continue;
            }
            onState(ERROR,&session,0);
            LogE("server recv err!\n");
            close(socket);
            socket = -1;
            break;
        } else if(n == 0) { //client端套接字被关闭
            LogE("xx server is closed\n");
            onState(CLOSE,&session,0);
            close(socket);
            socket = -1;
            break;
        } else { //读取成功
            /*对收到的数据进行处理*/
            LogI("server receive data:%d\n", n);
           // send(socket,buf,n,0);
            
            onMessage(&session,buf,n);
        }
    }
    LogI("DomainServerServlet::recvThreadLogic end\n");
}  
void DomainServerServlet::threadLogic()
{
    struct sockaddr_un server_addr;
    struct sockaddr_un conn_addr;
    socklen_t len;
    int conn_fd = -1;
    int server_fd = -1;

    /*创建unix域套接字*/
    server_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if(server_fd == -1)
    {
        LogE("server socket create error!\n");
        return;
    }

    /*指定server信息并绑定*/
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sun_family = AF_UNIX;
    snprintf(server_addr.sun_path, sizeof(server_addr.sun_path) - 1, url.c_str());

    unlink(url.c_str()); //如果系统中存在该路径，bind将失败，因此先调用unlink删除该路径。
    if(bind(server_fd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1)
    {
        LogE("server socket bind error! %s",strerror(errno));
        return;
    }

    /*监听server描述符*/
    if(listen(server_fd, 5) == -1)
    {
        LogE("server socket listen error!\n");
        return;
    }
    LogI("start to server %d",isRun);
    std::vector<std::thread*> threads;
    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(server_fd, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    isRun = true;
    while (isRun) {
        len = sizeof(conn_addr);
         
        /*接受并得到client信息*/
        conn_fd = accept(server_fd, (struct sockaddr *)&conn_addr, &len);
        if(conn_fd == -1)
        {
            //printf("accept failed %d",conn_fd);
            continue;
        }
        LogI("client is connected\n");
        auto client = new std::thread(&DomainServerServlet::recvthreadLogic,this,conn_fd);
        threads.push_back(client);
    }
    LogI("wait clients");
    for(auto i= threads.begin(); i != threads.end();++i) {
        (*i)->join();
    }
     
    /*关闭描述符*/
    LogI("server close");
    close(server_fd);
}
void DomainServerServlet::start()
{
    _thread = new std::thread(&DomainServerServlet::threadLogic,this);
}
void DomainServerServlet::wait()
{
    LogI("DomainServerServlet::wait\n");
    ((std::thread*)_thread)->join();
   
}
void DomainServerServlet::stop()
{
   isRun = false;
}
void DomainServerServlet::onUnload()
{
    stop();
    wait();
}
void DomainServerServlet::onLoad(INode *node)
{
    INode *child = node->childNode("connection");
    if (child) {
        name = child->getStr("name");
        url = child->getStr("url");
    }
    LogI("name:%s url:%s",name.c_str(),url.c_str());
    start();
}
void DomainServerServlet::onMessage(ISession *session,void* data,int size)
{
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
         for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void DomainServerServlet::onState(SERVLET_STATUS status,ISession *session,int code)
{
     auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
         for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}

DomainServerServlet::~DomainServerServlet()
{
     delete (std::thread*)_thread;
}

REG_PLUGIN(DomainServerServlet);
REG_PLUGIN(DomainClientServlet);
